package com.wg.neptune.netty.server;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Predicate;
import com.google.common.collect.Collections2;
import com.google.common.collect.Maps;
import com.wg.neptune.zookeeper.RegisterCenter;
import com.wg.neptune.model.ProviderService;
import com.wg.neptune.model.RpcRequest;
import com.wg.neptune.model.RpcResponse;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;

/**
 * Created by mc on 18/5/29.
 * netty事件处理类
 */
public class NettyServerHandler extends SimpleChannelInboundHandler<RpcRequest> {
    /*
     * 收到消息时，返回信息
     */

    //服务端限流
    private static final Map<String, Semaphore> serviceKeySemaphoreMap = Maps.newConcurrentMap();

    private static final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class);
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest rpcRequest)
            throws Exception {
        // 收到消息直接打印输出
        System.out.println("服务端接受的消息 : " + rpcRequest.toString());
        if (ctx.channel().isWritable()) {
            //从服务调用对象里获取服务提供者信息
            ProviderService metaDataModel = rpcRequest.getProviderService();
            long consumeTimeOut = rpcRequest.getInvokeTimeout();
            final String methodName = rpcRequest.getInvokedMethodName();

            //根据方法名称定位到具体某一个服务提供者
            String serviceKey = metaDataModel.getServiceItf().getName();

            //获取注册中心服务
            RegisterCenter registerCenter = RegisterCenter.singleton();
            List<ProviderService> providerServices = registerCenter.getProviderServiceMap().get(serviceKey);

            Object result = null;

            try {
                ProviderService providerService = Collections2.filter(providerServices, new Predicate<ProviderService>() {
                    public boolean apply(ProviderService input) {
                        return StringUtils.equals(input.getServiceMethod().getName(), methodName);
                    }
                }).iterator().next();
                //利用反射发起服务调用
                Object serviceObject = providerService.getServiceObject();
                Method method = providerService.getServiceMethod();
                result = method.invoke(serviceObject, rpcRequest.getArgs());
            } catch (Exception e) {
                System.out.println(JSON.toJSONString(providerServices) + "  " + methodName+" "+e.getMessage());
                result = e;
            }

            //根据服务调用结果组装调用返回对象
            RpcResponse response = new RpcResponse();
            response.setInvokeTimeout(consumeTimeOut);
            response.setUniqueKey(rpcRequest.getUniqueKey());
            response.setResult(result);

            //将服务调用返回对象回写到消费端
            ctx.writeAndFlush(response);
        } else {
            logger.error("------------channel closed!---------------");
        }
    }


     /** 建立连接时，返回消息**/

    /*@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("连接的客户端地址:" + ctx.channel().remoteAddress());
        ctx.writeAndFlush("客户端"+ InetAddress.getLocalHost().getHostName() + "成功与服务端建立连接！ \n");
        super.channelActive(ctx);
    }*/
}
